package com.atbeijing.app


import com.atbeijing.bean.StartUpLog
import com.atbeijing.constants.GmallConstants
import com.atbeijing.handler.D1
import com.atbeijing.utils.MyKafkaUtil
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.phoenix.spark._


object MyDauApp {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("aaa")
    val ssc = new StreamingContext(conf,Seconds(5))


    val kafkaDS: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_STARTUP,ssc)

    //组合计算逻辑,在行动算子运行前
    //转换样例类
    val objDS: DStream[StartUpLog] = D1.transformCase(kafkaDS)
    objDS.cache()
    objDS.count().print()

    //批次间去重
    val rDS: DStream[StartUpLog] = D1.filterByRedis(ssc,objDS)
    rDS.cache()
    rDS.count().print()

    //批次内去重
    val gDS: DStream[StartUpLog] = D1.groupDS(rDS)
    gDS.cache()
    gDS.count().print()


    //写入redis
    D1.writeRedis(gDS)

    //写入hbase
    gDS.foreachRDD(rdd =>{
      rdd.saveToPhoenix("GMALL2021_DAU",
        Seq("MID", "UID", "APPID", "AREA", "OS", "CH", "TYPE", "VS", "LOGDATE", "LOGHOUR", "TS"),
        HBaseConfiguration.create,
        Some("hadoop202,hadoop203,hadoop204:2181"))
    })

    //持续运行
    ssc.start()
    ssc.awaitTermination()

  }

}
